Bucketing

ETL pipelines regularly join large datasets on common keys (e.g. joining transactions with users on user_id). By default, every single join triggers an expensive network shuffle. If you run this join daily, you pay this heavy network shuffle cost daily.

graph TD
    subgraph Partitioning["Partitioning (partitionBy) - Creates folders"]
        direction TB
        F_Eng["/department=Engineering/"] --> File1["part-001.parquet"]
        F_Mkt["/department=Marketing/"] --> File2["part-002.parquet"]
    end
    subgraph Bucketing["Bucketing (bucketBy) - Creates locked file counts"]
        direction TB
        F_Table["/bucketed_transactions/"] --> B0["bucket_0.parquet (Hash keys 0, 4, 8)"]
        F_Table --> B1["bucket_1.parquet (Hash keys 1, 5, 9)"]
    end
    style Partitioning fill:#eff6ff,stroke:#2563eb,stroke-width:2px;
    style Bucketing fill:#f0fdf4,stroke:#16a34a,stroke-width:2px;

Bucketing is an optimization technique that pre-shuffles and pre-sorts your data at storage time. By saving your data on disk in pre-partitioned "buckets" based on key hashes, Spark completely skips the shuffle phase during future joins, reducing execution times from hours to minutes!

Partitioning vs. Bucketing

PySpark Code Example: Creating & Querying Bucketed Tables

Bucketing requires saving your data as persistent Spark Catalog Tables (e.g. using Hive Metastore or local catalog paths) rather than raw files, so that Spark can store and read the bucketing metadata:

from pyspark.sql import SparkSession

# 1. Setup Spark enabling local catalog configurations
spark = SparkSession.builder \
    .appName("Bucketing Tables") \
    .master("local[*]") \
    .getOrCreate()

# 2. Large Transactions DataFrame (Fact Table)
tx_data = [(101, 1, 500.0), (102, 2, 20.0), (103, 1, 150.0), (104, 3, 80.0)]
tx_df = spark.createDataFrame(tx_data, ["tx_id", "user_id", "amount"])

# 3. Large Users DataFrame (Dimension Table)
users_data = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
users_df = spark.createDataFrame(users_data, ["user_id", "user_name"])

# 4. Save both DataFrames as bucketed and sorted tables
# We choose 4 buckets, bucketed and sorted by 'user id'
tx_df.write \
    .format("parquet") \
    .mode("overwrite") \
    .bucketBy(4, "user_id") \
    .sortBy("user_id") \
    .saveAsTable("bucketed_transactions")

users_df.write \
    .format("parquet") \
    .mode("overwrite") \
    .bucketBy(4, "user_id") \
    .sortBy("user_id") \
    .saveAsTable("bucketed_users")

# 5. Read the tables from the Catalog
# Spark automatically reads the bucketing metadata from the catalog!
b_tx_df = spark.table("bucketed_transactions")
b_users_df = spark.table("bucketed_users")

# 6. Join the bucketed tables
# Because both tables are pre-shuffled into 4 buckets and pre-sorted by 'user id',
# Spark executes a high-speed join WITHOUT shuffles!
joined_df = b_tx_df.join(b_users_df, "user_id", "inner")
joined_df.show()

# 7. Check the physical plan
# You will see 'SortMergeJoin' but NO 'Exchange' (shuffle) steps in the plan!
joined_df.explain()